package com.arch.flink.biz.orderdemo;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.streaming.api.functions.co.RichCoFlatMapFunction;
import org.apache.flink.util.Collector;

/**
 * @author pizhihui
 * @date 2020-12-12
 */
public class OrderInfoDemo {


    private static final String ORDER_ADDRESS_FILE = "/Users/pizhihui/ProjectsIdea/code-bigdata/code-flink/src/input/order_address.txt";
    private static final String ORDER_DETAIL_FILE = "/Users/pizhihui/ProjectsIdea/code-bigdata/code-flink/src/input/order_detail.txt";


    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());


        DataStreamSource<String> addressSource = env.addSource(new OrderInfoSource(ORDER_ADDRESS_FILE));
        DataStreamSource<String> detailSource = env.addSource(new OrderInfoSource(ORDER_DETAIL_FILE));

        KeyedStream<OrderDetailBean, Long> detailStream = detailSource.map(new MapFunction<String, OrderDetailBean>() {
            @Override
            public OrderDetailBean map(String value) throws Exception {
                return OrderDetailBean.string2OrderDetail(value);
            }
        }).keyBy(OrderDetailBean::getOrderId);

        KeyedStream<OrderAddressBean, Long> addressStream = addressSource.map(new MapFunction<String, OrderAddressBean>() {
            @Override
            public OrderAddressBean map(String value) throws Exception {
                return OrderAddressBean.string2OrderAddress(value);
            }
        }).keyBy(OrderAddressBean::getOrderId);


        // 这里可以使用功能 flatmap,可以使用 process
        detailStream.connect(addressStream)
                //.process(new OrderTwoStreamProces())
                .flatMap(new OrderTwoStreamFlatMap())
                .print();

        env.execute(OrderInfoDemo.class.getSimpleName());

    }


    public static class OrderTwoStreamFlatMap extends RichCoFlatMapFunction<OrderDetailBean, OrderAddressBean, Tuple2<OrderDetailBean, OrderAddressBean>> {

        private ValueState<OrderDetailBean> detailValueState;
        private ValueState<OrderAddressBean> addressValueState;

        @Override
        public void open(Configuration parameters) throws Exception {
            detailValueState = getRuntimeContext().getState(
                    new ValueStateDescriptor<OrderDetailBean>("detail", OrderDetailBean.class));

            addressValueState = getRuntimeContext().getState(
                    new ValueStateDescriptor<OrderAddressBean>("address", OrderAddressBean.class));
        }

        @Override
        public void flatMap1(OrderDetailBean value, Collector<Tuple2<OrderDetailBean, OrderAddressBean>> out) throws Exception {
            OrderAddressBean addressBean = addressValueState.value();
            // addressBean不为空,表示addressBean的流来了
            if (null != addressValueState) {
                addressValueState.clear();
                out.collect(Tuple2.of(value, addressBean));
            } else {
                // addressBean流还没来,就将 detail 放入 state 中.
                detailValueState.update(value);
            }
        }

        @Override
        public void flatMap2(OrderAddressBean value, Collector<Tuple2<OrderDetailBean, OrderAddressBean>> out) throws Exception {
            OrderDetailBean detailBean = detailValueState.value();
            if (null == detailBean) {
                addressValueState.update(value);
            } else {
                detailValueState.clear();
                out.collect(Tuple2.of(detailBean, value));
            }
        }
    }

    public static class OrderTwoStreamProces extends CoProcessFunction<OrderDetailBean, OrderAddressBean, Tuple2<OrderDetailBean, OrderAddressBean>> {

        @Override
        public void processElement1(OrderDetailBean value, Context ctx, Collector<Tuple2<OrderDetailBean, OrderAddressBean>> out) throws Exception {

        }
        @Override
        public void processElement2(OrderAddressBean value, Context ctx, Collector<Tuple2<OrderDetailBean, OrderAddressBean>> out) throws Exception {

        }
        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<Tuple2<OrderDetailBean, OrderAddressBean>> out) throws Exception {
            super.onTimer(timestamp, ctx, out);
        }
    }

}
